package a.cb;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 表的标识和表的类型
 * Identifie =  Catalog名 +  数据库名 + TableName
 * <p>
 * Catalog名：  可以指向外部设备，HBase等 ，如果表持久化则必须使用
 * <p>
 * <p>
 * 表分为永久表和临时表
 * 创建临时表
 * 1、虚拟表：基于常规表得到的
 * 2、常规表：
 */
public class TableIdentifierAndType {

    public static void main(String[] args) {
//        掺杂DataStreamAPI方式
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        TableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
//        createTemporaryTable ，临时表
//        createTable ，可以创建常规表
//        以下SourceTable就是常规表
//        SQL 方式进行
//        Kafka Source
        tableEnvironment.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10l)
                .build());

//        Table table = tableEnvironment.from("SourceTable");
//        table.execute().print();
//        得到虚拟的表
        Table table2 = tableEnvironment.sqlQuery("  select * from  SourceTable ");
        table2.execute().print();
    }

    /**
     * 创建TableEnvironment
     *
     * @param args
     */
    public static void main1(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment environment = TableEnvironment.create(settings);
        environment.getConfig().getConfiguration().setString("parallelism.default", "1");
//        读取数据源
//
        environment.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10l)
                .build());
        Table table = environment.from("SourceTable");
        table.execute().print();


    }
}
